package com.huawei

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
 * args(0) : HDFS数据文件地址
 */
        object DataStatistics {
private val schemaString = "id,gender,height"

        def main(args: Array[String]): Unit = {
        if (args.length < 1) {
        //未输入数据源路径参数，则报错退出
        println("Usage:dataStatistics filePath")
        System.exit(1)
        }
        //配置工程
        val conf = new SparkConf().setAppName("dataStatistics")
        val sc = new SparkContext(conf)
        //屏蔽警告等级以下的信息
        sc.setLogLevel("WARN")
        //读取数据源路径
        val peopleDateRdd = sc.textFile(args(0).trim);
        val sqlCtx = new SQLContext(sc)

        val schemaArr = schemaString.split(",")
        //设置数据源的schema信息
        val schema = StructType(schemaArr.map(fieldName => StructField(fieldName,StringType,true)))
        //读取数据并转化为特定DataFrame格式
        val rowRdd : RDD[Row] = peopleDateRdd
        .map(_.split(","))
        .map(eachRow => Row(eachRow(0),eachRow(1),eachRow(2)))
        val peopleDF = sqlCtx.createDataFrame(rowRdd,schema)

        //将数据源缓存在内存中,会根据数据量节省很多时间
        peopleDF.persist(StorageLevel.MEMORY_ONLY_SER)
        peopleDF.createOrReplaceTempView("people")

        //获取男性身高超过180cm的数据
        val higherMale180 = sqlCtx.sql("select * from people where height > 180 and gender = 'M'")
        println("Men whose height are more than 180: " + higherMale180.count())

        //获取男性身高超过210cm的数据，只显示前10条
        println("Men whose height is more than 210")
        peopleDF.filter(peopleDF("gender").equalTo("M")).filter(peopleDF("height") > 210).show(10)

        //将身高由高到低排序，获取前10人的数据
        println("Sorted the people by height in descend order,Show top 10 people")
        peopleDF.sort(peopleDF("height").desc).take(10).foreach { println }

        //获取男性平均身高
        println("The Average height for Men")
        peopleDF.filter(peopleDF("gender").equalTo("M")).agg(Map("height" -> "avg")).show()

        //获取女性最高身高
        println("The Max height for Women:")
        peopleDF.filter(peopleDF("gender").equalTo("F")).agg("height" -> "max").show()

        println("All the statistics actions are finished on structured People data.")
        }
        }